I removed the null check and did it in update. e.g. MaxAggregate.update

Was this needed for correctness? I'm not sure I understand why you'd need to move it. The NullSafeAggregator was intended to avoid needing to handle null in the update methods. It also used to keep isNull set correctly so that any aggregate that is null would stop calling eval and updating.

If one data file evaluates null, I think we still want to evaluate the rest of the data files. For example,

CREATE TABLE test (id LONG, data INT) USING iceberg PARTITIONED BY (id);

INSERT INTO TABLE test VALUES (1, null), (1, null), (2, 33), (2, 44), (3, 55), (3, 66);

SELECT max(data) FROM test;

For max(data), the first data file evaluates null, I think we still want to evaluate the rest of the data files to get the max value 66 for max(data).

I am not sure I understand the purpose of isNull in this class then. Looks like we init it and never change?

I see. In that case, I think we need to change isNull to hasValue and return a boolean from update(R).

The intent here was to signal when there is not enough information to produce a value. When there isn't, then the result value should be null, and we can skip pulling values out of rows or data files because we don't have enough information.

For example, if we are processing 3 Parquet files and 1 Avro file, the Avro file may not have a max value. Rather than giving a partial max from the 3 Parquet files, we need update(avroFile) to return hasValue = false so that we stop aggregating.

You're right that this needs to change from my original version, which assumed any null value signaled that there was no maximum. If we know that a file contains only null values, then we can skip it even if it doesn't have an upper bound. Similarly, if we get a null value from a row then we can skip it.

I changed isNull to hasValue. I have also added a flag canPushDown in BoundAggregate to indicate if this aggregate can be pushed down. I think I need a way to differentiate the null: is the null due to stats not available (e.g. complex type) or due to the value is null, so I added this flag.

We typically separate words with - rather than _. I think this should also match the other property. How about spark.sql.iceberg.aggregate-push-down-enabled?

+1

Fixed. Thanks

Can you make this exception more specific? What might be thrown here?

Fixed. Thanks

This shouldn't swallow the exception by only printing the message. Instead, this should pass the exception to the logger so that it gets printed with the full stack trace, suppressed exceptions, and causes.

Fixed. Thanks

This is going to read all table metadata, which could be really large. Instead, I think this should use scan planning to get the data files. That will allow this to apply filters and skip a lot of data, and it would also parallelize manifest scanning using a ParallelIterator. You'd need to request stats, or else the tasks will be returned without them copied.

Changed to use scan planning to get the data files. Please take a look to see if it's OK.

I think it would be good to check whether the aggregates are non-null and only return if they are valid. Otherwise, this could return different results depending on whether stats are present in the file metadata. To avoid that, we can just detect whether we have a result and abort pushdown if we don't.

I have added the null check, but after a second thought, I removed it.

If one of the column has all null values, then the max or min is also null. We probably still want to push down the aggregate.

I have checked the Metrics mode to disable push down if the mode doesn't have stats. I have also disabled push down for complex types. I am wondering if it's safe without the null check here. If not, I will put back.

@huaxingao, I think the fix is to have a flag in the aggregator that can return whether or not the value is valid. That's what I wanted to use null for here, but you're right that there are cases where the aggregate is value and that value is null because there are no non-null values.

If we keep track of isValid in each aggregator, then the AggregateEvaluator can have a similar method to return whether all aggregates are valid. The we would just abort the aggregation if any value is not known. We can also have an override flag for when you want the closest answer, even if it isn't guaranteed to be correct.

FYI @aokolnychyi

OK thanks

Iceberg does not use get in method names. There's probably a better name here, like readSnapshot().

Fixed, thanks

Log messages should be more direct. In this case, the main information is that the aggregate pushdown is skipped. The reason why is secondary, but important. Rather than making this a statement that needs to be interpreted ("X is not possible" -> "Iceberg didn't do X"), this should be "Skipped aggregate pushdown: detected row level deletes".

In addition, I think that there are cases where you'd still want an answer from metadata. First, there may not be any matching delete files, so it could be safe. Second, it may be better to get an approximate answer. I think this should be handled using another setting that enables/disables aggregate pushdown when deletes are present. And for the first case, pushdown should only fail if there were delete files returned for at least one FileScanTask (as returned by planFiles called above).

I also support the idea of checking if any matching tasks have deletes and using that instead of relying on generic snapshot metadata.

I have changed the code to check the deletes in the tasks and abort the push down if deletes are present.

I also agree it may be better to introduce another setting to get an approximate number if there are deletes. Probably we can do this in a follow up PR.

Error message: don't use "yet". It is simply not supported.

It should be possible to do this, but I understand skipping it in the first PR.

Fixed. thanks

nit: What about a direct import for Aggregator to shorten the lines?

ImmutableList.Builder<Aggregator<?>> aggregatorsBuilder = ImmutableList.builder();

I am not sure I understand the purpose of isNull in this class then. Looks like we init it and never change?

+1

I'd consider adding another method to SparkAggregates to convert an entire Aggregation. That way, we will be able to simplify this block.

Changed. Thanks

I think that this test depends on the default metrics mode for string, not on the type being a string itself. If the metrics mode were 'full' so that values aren't truncated, then it would work. You may want to set the metrics mode explicitly for this test and test the case where the metrics are not truncated.

I actually set the metrics to full in the end of the test and tested pushed down

I'm glad to hear that there's a test for full, but the correctness of this test shouldn't rely on the default metrics mode. I think it should explicitly set the string column's mode to truncate.

It's the explain string that needs to be lower cased in these tests as well.

Why do we have to add this if we already have includeColumnStats defined in Scan?
I think we should be able to use that.

Right. Sorry I didn't notice there is already an existing method.

No problem at all. I forgot about that one initially too.